package com.lagou.rpc.consumer;

import com.lagou.rpc.api.IUserService;
import com.lagou.rpc.consumer.client.RpcClient;
import com.lagou.rpc.consumer.proxy.RpcClientProxy1;
import com.lagou.rpc.pojo.User;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 测试类
 */
public class ClientBootStrap {
    private static int flag = 0;
    public static void main(String[] args) throws InterruptedException {

//        IUserService userService = (IUserService) RpcClientProxy.createProxy(IUserService.class);
//        User user = userService.getById(1);
//        System.out.println(user);
//        System.out.println("-------------------------------------------");
//        IUserService userService1 = (IUserService) RpcClientProxy.createProxy(IUserService.class);
//        User user1 = userService1.getById(1);
//        System.out.println(user1);


        ZkClient zkClient = new ZkClient("127.0.0.1:2181");
        System.out.println("会话被创建了..");

        String basePath = "/zk-rpc";

        Map<String, RpcClient> map = new HashMap<>();

        // 获取子节点列表
        List<String> children = zkClient.getChildren(basePath);
        System.out.println(children);

        for (int i = 0; i < children.size(); i++) {
            String node =  children.get(i);
            String[] split = node.split(":");
            int port = Integer.parseInt(split[1]);
            RpcClient rpcClient = new RpcClient(split[0],port);
            System.out.println("port:" + port);
            map.put(node,rpcClient);
        }

//        System.out.println("--------------------test-----------------------");
//        RpcClient rpcClient = map.get("127.0.0.1:8891");
//        IUserService userService = (IUserService) RpcClientProxy1.createProxy(IUserService.class,rpcClient);
//        User user = userService.getById(2);
//        System.out.println(user);
//        System.out.println("--------------------test-----------------------");

        // 注册监听
        zkClient.subscribeChildChanges(basePath, new IZkChildListener() {
            @Override
            public void handleChildChange(String parentPath, List<String> list) throws Exception {
                System.out.println(parentPath + "的子节点列表发生了变化,变化后的子节点列表为"+ list);

                for(Map.Entry<String, RpcClient> entry:map.entrySet()){

                    if(!list.contains(entry.getKey())) {
                        RpcClient rpcClient = map.get(entry.getKey());
                        rpcClient.close();
                        map.remove(entry.getKey());
                        System.out.println(entry.getKey()+"--->remove");
                    }
                }

                for(String node:list) {

                    if(!map.containsKey(node)){
                        String[] split = node.split(":");
                        int port = Integer.parseInt(split[1]);
                        RpcClient rpcClient = new RpcClient(split[0],port);
                        System.out.println("port:" + port);
                        map.put(node,rpcClient);
                    }
                }
            }
        });

        ScheduledExecutorService pool = Executors.newScheduledThreadPool(15);

        /**
         * 这个执行周期是固定，不管任务执行多长时间，还是每过5秒中就会产生一个新的任务
         */
        pool.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                //这个业务逻辑需要很长的时间，定时任务去统计一张数据上亿的表，财务财务信息，需要30min
                System.out.println("------------------每5秒更新一次响应时间---------------");
                for (int i = 0; i < children.size(); i++) {
                    long currentTime = System.currentTimeMillis();
                    String node =  children.get(i);
                    String data = zkClient.readData(basePath + "/" + node);
                    System.out.println(data);
                    if(data == null) {
                        // 更新节点内容
                        zkClient.writeData(basePath + "/" + node, "999:" + 0L);
                        System.out.println(basePath + node + "的数据为：" + "999:" + 0L);
                        continue;
                    }
                    String[] split = data.split(":");
                    long useTime = Long.parseLong(split[0]);
                    System.out.println("currentTime:" + currentTime);
                    System.out.println("gap:" + (currentTime - Long.parseLong(split[1])));
                    if ((currentTime - Long.parseLong(split[1])) > 5000L) {
                        // 更新节点内容
                        zkClient.writeData(basePath + "/" + node,useTime + ":" + 0L);
                        System.out.println(basePath + node + "的数据为：" + useTime + ":" + 0L);
                    }
                }
            }
        },1,5,TimeUnit.SECONDS);



        /**
         * 这个执行周期是固定，不管任务执行多长时间，还是每过3秒中就会产生一个新的任务
         */
        pool.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                System.out.println("------------------每10秒发送一次请求---------------");
                long tempTime = 5000L;
                int port = 8891;
                if(flag%2 == 0) {
                    port = 8899;
                }
                flag++;
                String goodNode = "127.0.0.1:"+port;
                for (int i = 0; i < children.size(); i++) {
                    // 读取节点内容
                    String node =  children.get(i);
                    String data = zkClient.readData(basePath + "/" + node);
                    System.out.println(node + ":" + data);
                    if(data == null) { continue; }
                    String[] split = data.split(":");
                    long useTime = Long.parseLong(split[0]);
                    if (Long.parseLong(split[1]) == 0L) {
                        continue;
                    }
                    if(useTime < tempTime) {
                        goodNode = node;
                        tempTime = useTime;
                    }
                }

                //String goodNode = "127.0.0.1:8891";
                long startTime=System.currentTimeMillis();
                IUserService userService = (IUserService) RpcClientProxy1.createProxy(IUserService.class,goodNode);
                User user = userService.getById(2);
                long endTime=System.currentTimeMillis();
                long useTime = endTime - startTime;
                System.out.println(goodNode + "RPC接口调用时间：" + useTime + "ms");
                System.out.println(user);
                // 更新节点内容
                String data = useTime + ":" + endTime;
                zkClient.writeData(basePath + "/" + goodNode,data);
                System.out.println(basePath + "/" + goodNode + "的数据为：" + useTime + ":" + endTime);
            }
        },2,10, TimeUnit.SECONDS);

    }
}
